package com.ksyun.campus.client.util;

import com.alibaba.fastjson.JSONObject;
import com.ksyun.campus.client.domain.ClusterInfo;
import com.ksyun.campus.client.param.DataServerParse;
import com.ksyun.campus.client.param.MetaServerParse;
import org.I0Itec.zkclient.ZkClient;

import javax.annotation.PostConstruct;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;


public class ZkUtil {
    /**
     * 集群信息
     */
    private ClusterInfo clusterInfo;
    @PostConstruct
    public void postCons() throws Exception {
        // 获取配置文件
        Properties properties = new Properties();
        String projectRoot = System.getProperty("user.dir");
        try (FileInputStream inputStream = new FileInputStream(projectRoot+ "easyClient/src/main/resources/application.yml")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
        String zooKeeperAddress = properties.getProperty("zookeeper.address");
        int timeOutTime = Integer.parseInt(properties.getProperty("zookeeper.timeout"));
        String metaServerRoot = properties.getProperty("metaserver.root-name","/metaserver");
        String dataServerRoot = properties.getProperty("dataserver.root-name","/dataserver");

        // todo 初始化，与zk建立连接，注册监听路径，当配置有变化随时更新
        // 识别metaserver maser
        ZkClient zkClient = new ZkClient(zooKeeperAddress,timeOutTime);

        clusterInfo = new ClusterInfo();

        // 设置metaserver, metaServerNodes监听节点下为有序节点, 节点小的作为主metaServer, 大的作为从metaServer
        if (zkClient.exists(metaServerRoot)){
            List<String> metaServerNodes = zkClient.getChildren(metaServerRoot).stream().sorted().collect(Collectors.toList());
            if (metaServerNodes.size()>0){
                MetaServerParse metaServerParse = parseObjectFrom(zkClient, metaServerNodes.get(0), MetaServerParse.class);
                ClusterInfo.MetaServerMsg masterMetaServer = clusterInfo.new MetaServerMsg();
                masterMetaServer.setHost(metaServerParse.getIp());
                masterMetaServer.setPort(Integer.parseInt(metaServerParse.getPort()));
                clusterInfo.setMasterMetaServer(masterMetaServer);
            }
            if (metaServerNodes.size()>1){
                MetaServerParse metaServerParse = parseObjectFrom(zkClient, metaServerNodes.get(1), MetaServerParse.class);
                ClusterInfo.MetaServerMsg slaveMetaServer = clusterInfo.new MetaServerMsg();
                slaveMetaServer.setHost(metaServerParse.getIp());
                slaveMetaServer.setPort(Integer.parseInt(metaServerParse.getPort()));
                clusterInfo.setSlaveMetaServer(slaveMetaServer);
            }
        }
        if (zkClient.exists(dataServerRoot)){
            // 设置dataserver
            List<String> dataServerNodes = zkClient.getChildren(dataServerRoot);
            //线程安全的CopyOnWriteArrayList
            List<ClusterInfo.DataServerMsg> dataServerMsgList = new CopyOnWriteArrayList<>();
            for (String dsNode: dataServerNodes) {
                DataServerParse dataServerParse = parseObjectFrom(zkClient, dsNode, DataServerParse.class);
                ClusterInfo.DataServerMsg dataServerMsg = clusterInfo.new DataServerMsg();
                dataServerMsg.setHost(dataServerParse.getIp());
                dataServerMsg.setPort(Integer.parseInt(dataServerParse.getPort()));
                dataServerMsg.setCapacity(dataServerParse.getCapacity());
                dataServerMsg.setUseCapacity(dataServerParse.getUsedCapacity());
                dataServerMsg.setFileTotal(dataServerParse.getAllReplicaDataMap().size());

                dataServerMsgList.add(dataServerMsg);
            }
            clusterInfo.setDataServer(dataServerMsgList);
        }

        System.out.println(clusterInfo);
    }

    private <T> T parseObjectFrom(ZkClient zkClient, String node,  Class<T> responseType){
        String dataServerInfoJsonStr = (String) zkClient.readData(node);
        try {
            T result = JSONObject.parseObject(dataServerInfoJsonStr, responseType);
            return result;
        }catch (RuntimeException e) {
//            throw new JSONException("can not cast to JSONObject.", e);
            return null;
        }
    }

    public ClusterInfo getClusterInfo() {
        return clusterInfo;
    }
}
